/*
   Copyright 2009 Abdulla Abdurakhmanov (abdulla@latestbit.com)
   Original sources are available at www.latestbit.com

   Licensed under the Apache License, Version 2.0 (the "License");
   you may not use this file except in compliance with the License.
   You may obtain a copy of the License at

       http://www.apache.org/licenses/LICENSE-2.0

   Unless required by applicable law or agreed to in writing, software
   distributed under the License is distributed on an "AS IS" BASIS,
   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   See the License for the specific language governing permissions and
   limitations under the License.
*/

package org.latestbit.smartobjects.network.tcp;

import java.io.IOException;
import java.nio.channels.CancelledKeyException;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectableChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.util.concurrent.locks.ReentrantLock;

public class SocketMultiplexer implements Runnable {
	private static final long CHECK_KEEP_ALIVE_INTERVAL = 1000;
	
	private Selector selector;
	private volatile boolean stopFlag = false;
	private final int MAX_SELECTED_CHANNELS = 512;
	private int channelsCnt;
	private Thread thread;
	private Object groupName;
	private ReentrantLock selectorLock = new ReentrantLock();
	private ReentrantLock multiplexerLock = new ReentrantLock();	
	
	public SocketMultiplexer(String groupName) throws IOException {
		selector = Selector.open();
		this.groupName = groupName;
		thread = new Thread ( this, "SOCK_MTPX["+this.groupName+"]" );
		thread.setDaemon(true);
		thread.start();
	}
	
	public SelectionKey addChannel ( SelectableChannel channel, 
							int ops, 
							ISocketMultiplexerListener listener ) 
								throws ClosedChannelException, IOException {
		multiplexerLock.lock();
		try {
			selector.wakeup();
			SelectionKey key = channel.register( selector, ops, listener );
			channelsCnt++;
			return key;
		}
		finally {
			multiplexerLock.unlock();
		}
	}
	
	public void removeChannel( SelectionKey key ) {
		multiplexerLock.lock();
		try {
			selector.wakeup();
			key.cancel();
			channelsCnt--;
		}
		finally {
			multiplexerLock.unlock();
		}
	}
	
	public void finalize() {
		destroy();
	}
	
	public synchronized void destroy() {
		multiplexerLock.lock();
		stopFlag = true;
		if(selector!=null && selector.isOpen()) {			
			try { selector.wakeup(); } catch (Exception ex) {};				
		}
		multiplexerLock.unlock();
		
		selectorLock.lock();
		try {
			if(selector!=null && selector.isOpen()) {
				try {
					selector.close();				
				}
				catch(Exception ex) {};
				selector=null;
			}
		}
		finally {
			selectorLock.unlock();
		}		

		if(thread.isAlive())
			try { thread.join(); } catch(Exception ex) {};		
	}

	@Override
	public void run() {
		try {			
			while(!stopFlag) {				
				try {					
					int cnt = 0;
					selectorLock.lock();
					try {
						cnt = selector.select( CHECK_KEEP_ALIVE_INTERVAL );
							
					}
					finally {
						selectorLock.unlock();
					}
					
					multiplexerLock.lock();
					try {
						if(stopFlag)
							break;
					}
					finally {
						multiplexerLock.unlock();
					}
					
					if(cnt > 0 ) {
						if(selector.isOpen()) {
							for ( SelectionKey key : selector.selectedKeys() ) {
								ISocketMultiplexerListener listener = (ISocketMultiplexerListener)key.attachment();
								listener.onSelectedEvent(key);
							}
						}
					}
					
					selectorLock.lock();
					try {
						if(selector!=null && selector.isOpen()) {
							for( SelectionKey key : selector.keys() ) {
								ISocketMultiplexerListener listener = (ISocketMultiplexerListener)key.attachment();
								listener.onCheckKeepAlive(key);
							}
							
							selector.selectedKeys().clear();
						}
					}
					finally {
						selectorLock.unlock();
					}
				} 
				catch (IOException e) {
					e.printStackTrace();
				}				
			}
		}
		catch(CancelledKeyException ex) {
			ex.printStackTrace();
		}
	}
	
	public boolean isFull() {
		if ( channelsCnt >= MAX_SELECTED_CHANNELS )
			return true;
		else
			return false;
	}

	public void printSelectorStatus() {
		System.out.println("Selector :"+selector+" ("+channelsCnt+")");
		for(SelectionKey key: selector.keys()) {
			System.out.println("Key :"+key.channel());
		}
	}
}
